build

Greptile Summary

This PR adds an internal config spark.rapids.sql.rowToColumnar.retry.enabled (defaulting to false) that disables the per-row retry wrapper introduced in #13842 for RowToColumnarIterator, addressing a reported performance regression. When disabled, each row is converted directly without a withRetryNoSplit/withRestoreOnRetry wrapper, removing per-row overhead at the cost of propagating host OOM exceptions directly to the task level. The builders.tryBuild(rowCount) call remains wrapped in withRetryNoSplit in both paths, so GPU OOM at the build step is still retried.

• RowToColumnarIterator gains an enableRetry: Boolean = false constructor parameter that gates the two conversion branches.
• Both production call sites (GpuRowToColumnarExec and ParquetCachedBatchSerializer) now read rapidsConf.isR2cRetryEnabled and forward it to the iterator.
• Four new tests are added: CPU OOM without retry (propagates), CPU OOM with retry (recovers), multi-batch retry correctness, and a single-batch recovery case using a custom rowThatFailsOnceWithCpuRetryOOM helper that throws exactly once.
• Existing retry tests are updated to pass enableRetry = true explicitly to preserve original behaviour coverage.

Confidence Score: 5/5

• Safe to merge — the change is gated behind an internal config that defaults to the new (retry-disabled) behaviour, and all production call sites have been updated.
• The implementation is logically correct: the non-retry path removes only the per-row OOM wrapper while keeping the final builders.tryBuild retry intact. Both production callers read and forward the new config. The new tests directly validate all four scenarios (retry on/off × OOM recovery/propagation). No pre-existing retry tests have been broken; they were updated to pass enableRetry = true explicitly. The config is marked .internal() so no user-facing documentation update is strictly required.
• No files require special attention.

Important Files Changed

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A["RowToColumnarIterator.buildBatch()"] --> B{enableRetry?}

    B -- true --> C["Create RetryableRowConverter\n(with UnsafeProjection)"]
    C --> D["converter.attempt(rowIter.next())"]
    D --> E["withRetryNoSplit {\n  withRestoreOnRetry(converter) {\n    converters.convert(currentRow, builders)\n  }\n}"]
    E -- OOM --> F["converter.restore()\nRollback builders + copy row"]
    F --> E
    E -- success --> G["byteCount += bytesWritten\nrowCount += 1"]

    B -- false --> H["val row = rowIter.next()"]
    H --> I["converters.convert(row, builders)"]
    I -- OOM --> J["CpuRetryOOM propagates\n(task may fail)"]
    I -- success --> K["byteCount += bytes\nrowCount += 1"]

    G --> L{more rows\nwithin target?}
    K --> L
    L -- yes --> D
    L -- yes --> H
    L -- no --> M["withRetryNoSplit {\n  builders.tryBuild(rowCount)\n}"]
    M --> N["return ColumnarBatch"]

_{Last reviewed commit: a3f34e9}